package com.consumer.kafka.starter.configuration;

import com.consumer.kafka.starter.core.KafkaSource;
import com.consumer.kafka.starter.core.KafkaMultiSourceProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * @author: xinyue
 * @date: 2021/6/23 14:20
 */
@Slf4j
@Component(value = "kafkaMultiSourceRegister")
public class KafkaMultiSourceRegister implements InitializingBean {

    @Autowired
    private KafkaMultiSourceProperties kafkaMultiSourceProperties;

    @Autowired
    private DefaultListableBeanFactory beanFactory;


    /**
     * 消费者配置
     */
    private Map<String, Object> consumerConfig(KafkaSource kafkaSource) {
        Map<String, Object> props = new HashMap<>(10);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSource.getBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaSource.getGroupId());
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, kafkaSource.getClientId());
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, kafkaSource.getHeartbeatInterval());
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, kafkaSource.getFetchMinSize());
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, kafkaSource.getFetchMaxWait());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaSource.getMaxPollRecords());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaSource.isEnableAutoCommit());
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaSource.getSessionTimeoutMs());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaSource.getAutoOffsetReset());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    /**
     * 自动注入
     *
     * @throws Exception 异常
     */
    @Override
    public void afterPropertiesSet() {
        if(CollectionUtils.isEmpty(kafkaMultiSourceProperties.getMultiSource())){
            return;
        }

        kafkaMultiSourceProperties.getMultiSource().forEach(kafkaSource -> {
            DefaultKafkaConsumerFactory<Object, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfig(kafkaSource));


            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory);
            factory.getContainerProperties().setPollTimeout(1500);
            if (!kafkaSource.isEnableAutoCommit()) {
                //配置手动提交offset
                factory.getContainerProperties().setAckMode((ContainerProperties.AckMode.MANUAL));
            }
            //注册bean实例
            beanFactory.registerSingleton(kafkaSource.getClientId() + "-ConsumerFactory", consumerFactory);
            beanFactory.registerSingleton(kafkaSource.getClientId() + "-KafkaListenerContainerFactory", factory);
            System.out.println("beanName:" + kafkaSource.getClientId() + "-ConsumerFactory");
            System.out.println("beanName:" + kafkaSource.getClientId() + "-KafkaListenerContainerFactory");
        });
    }
}
